package com.jcj.jcjmq.client;

import cn.kimmking.utils.HttpUtils;
import cn.kimmking.utils.ThreadUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.jcj.jcjmq.model.Message;
import com.jcj.jcjmq.model.Result;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;

@Slf4j
public class JcjBroker {
    public static String brokerUrl = "http://localhost:8765/jcjmq";

    @Getter
    private static JcjBroker defaultBroker = new JcjBroker();

    static {
        ThreadUtils.getDefault().init(1);
        ThreadUtils.getDefault().schedule(() -> {
            defaultBroker.getConsumers().forEach((topic, consumers) -> {
                consumers.forEach(
                        consumer -> {
                            Message<?> recv = consumer.recv(topic);
                            if (recv == null) {
                                return;
                            }
                            try {
                                consumer.getListener().onMessage(recv);
                                consumer.ack(topic, recv);
                            } catch (Exception e) {
                                log.error("onMessage error: " + e.getMessage());
                            }
                        }
                );
            });
        }, 100, 100);
    }

    public JcjProducer createProducer() {
        return new JcjProducer(this);
    }

    public JcjConsumer createConsumer(String topic) {
        JcjConsumer<?> consumer = new JcjConsumer<>(this);
        consumer.sub(topic);
        return consumer;
    }

    public boolean send(String topic, Message message) {
        log.info(" ==>> send message: " + JSON.toJSONString(message));
        Result<String> result = HttpUtils.httpPost(JSON.toJSONString(message),
                brokerUrl + "/send?topic=" + topic,
                new TypeReference<Result<String>>() {
                });
        log.info(" ==>> send result: " + result);

        return result.getCode() == 1;
    }

    public void sub(String topic, String id) {
        log.info("sub topic: " + topic + " consumerId: " + id);
        Result<String> stringResult = HttpUtils.httpGet(
                brokerUrl + "/sub?topic=" + topic + "&consumerId=" + id,
                new TypeReference<Result<String>>() {
                });
    }

    public <T> Message<T> recv(String topic, String id) {
        log.info(" ==>> recv topic: " + topic + " consumerId: " + id);
        Result<Message<String>> result = HttpUtils.httpGet(
                brokerUrl + "/recv?topic=" + topic + "&consumerId=" + id,
                new TypeReference<Result<Message<String>>>() {
                });
        log.info(" ==>> recv result: " + result);
        return (Message<T>) result.getData();

    }

    public void unsub(String topic, String id) {
        log.info("unsub topic: " + topic + " consumerId: " + id);
        Result<String> stringResult = HttpUtils.httpGet(
                brokerUrl + "/unsub?topic=" + topic + "&consumerId=" + id,
                new TypeReference<Result<String>>() {
                });
    }



    public boolean ack(String topic, String id, int offset) {
        log.info("ack topic,url=" + brokerUrl + "/ack?topic=" + topic + "&consumerId=" + id + "&offset=" + offset);
        Result<String> result = HttpUtils.httpGet(brokerUrl + "/ack?topic=" + topic + "&consumerId=" + id + "&offset=" + offset,
                new TypeReference<Result<String>>() {
                });
        log.info("ack result: " + result);
        return result.getCode() == 1;
    }

    @Getter
    private MultiValueMap<String, JcjConsumer<?>> consumers = new LinkedMultiValueMap();

//    public <T> void addListener(String topic, JcjListener<T> listener) {
//        consumers.add(topic, listener);
//    }

    public <T> void addConsumer(String topic, JcjConsumer<T> tJcjConsumer) {
        consumers.add(topic, tJcjConsumer);
    }
}

